package com.opdar.athena.registry;

import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.NetworkInterface;
import java.net.SocketException;
import java.util.Enumeration;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * Created by shiju on 2017/6/27.
 */
public class ZookeeperRegistry  implements Runnable, Watcher {
    private final String host;
    private String module = "DEFAULT";
    private String bind = "*.*.*.*";
    private String root = "/com/opdar/athena";
    private ZooKeeper zooKeeper;
    protected CountDownLatch countDownLatch = new CountDownLatch(1);
    private List<InetSocketAddress> addresses = new LinkedList<InetSocketAddress>();

    public ZookeeperRegistry(String host) {
        this("DEFAULT", host);
    }

    public ZookeeperRegistry(String module, String host) {
        this.host = host;
        try {
            this.module = module;
            connect(host);
            String[] paths = root.split("/");
            String _root = "";
            for (String c : paths) {
                if (c.length() == 0) continue;
                _root += "/" + c;
                ZookeeperUtils.createPath(zooKeeper, _root, "", CreateMode.PERSISTENT);
            }

        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    private void connect(String host) throws IOException, InterruptedException {
        zooKeeper = new ZooKeeper(host, 1000, new Watcher() {
            @Override
            public void process(WatchedEvent event) {
                if (Event.KeeperState.SyncConnected == event.getState()) {
                    countDownLatch.countDown();
                }else if (Event.KeeperState.Disconnected == event.getState()){
                    try {
                        zooKeeper.close();
                        zooKeeper = null;
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }else if (Event.KeeperState.Expired == event.getState()){
                    try {
                        connect(ZookeeperRegistry.this.host);
                    } catch (Exception e) {
                        e.printStackTrace();
                    }
                }
            }
        });
        countDownLatch.await();
        read();
    }

    public Stat setData(String path, byte[] data){
        try {
            return zooKeeper.setData(root + "/" + module + "/" + path,data,-1);
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        return null;
    }
    public byte[] getData(String path){
        try {
            return zooKeeper.getData(root + "/" + module + "/" + path,false,null);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    public Stat exists(String path,Watcher watcher) throws KeeperException, InterruptedException {
        return zooKeeper.exists(root + "/" + module+"/"+path, watcher);
    }

    public List<String> existsModule(Watcher watcher) throws KeeperException, InterruptedException {
        return zooKeeper.getChildren(root + "/" + module, watcher);
    }

    public ZooKeeper.States getState() {
        return zooKeeper.getState();
    }

    public ZooKeeper getZooKeeper() {
        return zooKeeper;
    }

    public void setBind(String bind) {
        this.bind = bind;
    }


    public void add(String value) {
        add(value,"".getBytes());
    }

    public void add(String value,byte[] data) {
        try {
            String[] address = value.split(":");
            InetSocketAddress _address = new InetSocketAddress(address[0], Integer.valueOf(address[1]));
            addresses.add(_address);
            ZookeeperUtils.createPath(zooKeeper, root + "/" + module, "", CreateMode.PERSISTENT);
            ZookeeperUtils.createPath(zooKeeper, root + "/" + module + "/" + value, data, CreateMode.EPHEMERAL);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void read() {
        try {
            if(zooKeeper == null)connect(ZookeeperRegistry.this.host);
            Stat stat = zooKeeper.exists(root + "/" + module, false);
            if (stat == null) {
                return;
            }
            List<String> values = zooKeeper.getChildren(root + "/" + module, this);
            addresses.clear();
            for (String c : values) {
                String[] address = c.split(":");
                addresses.add(new InetSocketAddress(address[0], Integer.valueOf(address[1])));
            }
        } catch (KeeperException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    public void stop() {
        try {
            zooKeeper.close();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void run() {
        while (true) {

        }
    }

    @Override
    public void process(WatchedEvent watchedEvent) {
        switch (watchedEvent.getType()) {
            case NodeChildrenChanged:
                read();
                break;
            default:
                //None...
                System.out.println(watchedEvent);
                break;
        }
    }

    public List<InetSocketAddress> getAddresses() {
        return addresses;
    }

    public String bindAddress() {
        try {
            Enumeration netInterfaces = NetworkInterface.getNetworkInterfaces();
            while (netInterfaces.hasMoreElements()) {
                NetworkInterface ni = (NetworkInterface) netInterfaces
                        .nextElement();
                Enumeration nii = ni.getInetAddresses();
                while (nii.hasMoreElements()) {
                    InetAddress ip = (InetAddress) nii.nextElement();
                    if (!ip.getHostAddress().contains(":")) {
                        String[] h = ip.getHostAddress().split("\\.");
                        String[] b = bind.split("\\.");
                        int i = 0;
                        for (i = 0; i < b.length; i++) {
                            String s = b[i];
                            if (s.equals("*")) continue;
                            if (!h[i].equals(s)) break;
                        }
                        if (i == 4) {
                            return ip.getHostAddress();
                        }
                    }
                }
            }
        } catch (SocketException e) {
            e.printStackTrace();
        }
        return "127.0.0.1";
    }

    public long size() {
        return addresses.size();
    }

}